/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.producer;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.Validators;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.hook.CheckForbiddenContext;
import com.alibaba.rocketmq.client.hook.CheckForbiddenHook;
import com.alibaba.rocketmq.client.hook.SendMessageContext;
import com.alibaba.rocketmq.client.hook.SendMessageHook;
import com.alibaba.rocketmq.client.impl.CommunicationMode;
import com.alibaba.rocketmq.client.impl.MQClientManager;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.producer.*;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.ServiceState;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.help.FAQUrl;
import com.alibaba.rocketmq.common.message.*;
import com.alibaba.rocketmq.common.protocol.ResponseCode;
import com.alibaba.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.EndTransactionRequestHeader;
import com.alibaba.rocketmq.common.protocol.header.SendMessageRequestHeader;
import com.alibaba.rocketmq.common.sysflag.MessageSysFlag;
import com.alibaba.rocketmq.remoting.RPCHook;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.common.RemotingUtil;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-24
 */
public class DefaultMQProducerImpl implements MQProducerInner {
	private final Logger log = ClientLogger.getLog();
	private final DefaultMQProducer defaultMQProducer;
	private final ConcurrentHashMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable = new ConcurrentHashMap<String, TopicPublishInfo>();
	protected BlockingQueue<Runnable> checkRequestQueue;
	protected ExecutorService checkExecutor;
	private ServiceState serviceState = ServiceState.CREATE_JUST;
	private MQClientInstance mQClientFactory;

	private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
	private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>();
	private final RPCHook rpcHook;

	public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
		this.defaultMQProducer = defaultMQProducer;
		this.rpcHook = rpcHook;
	}

	public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) {
		this(defaultMQProducer, null);
	}

	public boolean hasCheckForbiddenHook() {
		return !checkForbiddenHookList.isEmpty();
	}

	public void registerCheckForbiddenHook(CheckForbiddenHook checkForbiddenHook) {
		this.checkForbiddenHookList.add(checkForbiddenHook);
		log.info("register a new checkForbiddenHook. hookName={}, allHookSize={}", checkForbiddenHook.hookName(), checkForbiddenHookList.size());
	}

	public void executeCheckForbiddenHook(final CheckForbiddenContext context) throws MQClientException {
		if (hasCheckForbiddenHook()) {
			for (CheckForbiddenHook hook : checkForbiddenHookList) {
				hook.checkForbidden(context);
			}
		}
	}

	public void initTransactionEnv() {
		TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
		this.checkRequestQueue = new LinkedBlockingQueue<Runnable>(producer.getCheckRequestHoldMax());
		this.checkExecutor = new ThreadPoolExecutor(//
				producer.getCheckThreadPoolMinSize(), //
				producer.getCheckThreadPoolMaxSize(), //
				1000 * 60, //
				TimeUnit.MILLISECONDS, //
				this.checkRequestQueue);
	}

	public void destroyTransactionEnv() {
		this.checkExecutor.shutdown();
		this.checkRequestQueue.clear();
	}

	public boolean hasSendMessageHook() {
		return !this.sendMessageHookList.isEmpty();
	}

	public void registerSendMessageHook(final SendMessageHook hook) {
		this.sendMessageHookList.add(hook);
		log.info("register sendMessage Hook, {}", hook.hookName());
	}

	public void executeSendMessageHookBefore(final SendMessageContext context) {
		if (!this.sendMessageHookList.isEmpty()) {
			for (SendMessageHook hook : this.sendMessageHookList) {
				try {
					hook.sendMessageBefore(context);
				} catch (Throwable e) {
				}
			}
		}
	}

	public void executeSendMessageHookAfter(final SendMessageContext context) {
		if (!this.sendMessageHookList.isEmpty()) {
			for (SendMessageHook hook : this.sendMessageHookList) {
				try {
					hook.sendMessageAfter(context);
				} catch (Throwable e) {
				}
			}
		}
	}

	public void start() throws MQClientException {
		this.start(true);
	}

	public void start(final boolean startFactory) throws MQClientException {
		// 1、检查DefaultMQProducerImpl.ServiceState的状态（初始化状态为ServiceState.CREATE_JUST）；
		// 只有状态为CREATE_JUST时才启动该Producer；其他状态均不执行启动过程；
		switch (this.serviceState) {
			case CREATE_JUST :
				// 2、将DefaultMQProducerImpl.ServiceState置为start_failed,以免客户端同一个进程中重复启动；
				this.serviceState = ServiceState.START_FAILED;

				// 3、检查producerGroup是否合法：不能为空、长度不能大于255、不能等于"DEFAULT_PRODUCER"；若不合法则直接向应用层抛出MQClientException异常；
				this.checkConfig();

				// 4、若producerGroup不等于"CLIENT_INNER_PRODUCER"则设置Producer的实例名（instanceName）；
				// 调用java的ManagementFactory.getRuntimeMXBean()方法获取该进程的PID作为该Producer的实例名；
				if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
					this.defaultMQProducer.changeInstanceNameToPID();
				}

				// 5、构建该Producer的ClientID，等于IP地址@instanceName；
				// 6、创建MQClientInstance对象
				// 6.1）初始化ClientRemotingProcessor对象，处理接受的事件请求；
				// 6.2）初始化MQClientAPIImpl对象，在初始化过程中，初始化了MQClientAPIImpl.remotingClient:NettyRemotingClient对象，将ClientRemotingProcessor对象作为事件处理器注册到NettyRemotingClient对象中，处理的事件号有：CHECK_TRANSACTION_STATE、NOTIFY_CONSUMER_IDS_CHANGED、RESET_CONSUMER_CLIENT_OFFSET、GET_CONSUMER_STATUS_FROM_CLIENT、GET_CONSUMER_RUNNING_INFO、CONSUME_MESSAGE_DIRECTLY。
				// 6.4）初始化PullMessageService、RebalanceService、ConsumerStatsManager服务线程；PullMessageService服务线程是供DefaultMQPushConsumer端使用的，RebalanceService服务线程是供Consumser端使用的；
				this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

				// 7、将DefaultMQProducerImpl对象在MQClientInstance中注册
				// 以producerGroup为key值、DefaultMQProducerImpl对象为values值存入MQClientInstance.producerTable:ConcurrentHashMap<String/* group */, MQProducerInner>变量中，
				// 若在该变量中已存在该producerGroup的记录则向应用层抛出MQClientException异常；
				// 说明在一个客户端的一个进程下面启动多个Producer时producerGroup名字不能一样，否则无法启动；
				boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
				if (!registerOK) {
					this.serviceState = ServiceState.CREATE_JUST;
					throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup() + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null);
				}

				// 8、以主题名"TBW102"为key值，新初始化的TopicPublishInfo对象为value值存入DefaultMQProducerImpl.topicPublishInfoTable变量中；
				this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey()/* TBW102 */, new TopicPublishInfo());

				if (startFactory) {
					// ****************** 9、入参startFactory等于true，故调用MQClientInstance.start方法启动MQClientInstance对象；*********************
					mQClientFactory.start();
				}

				log.info("the producer [{}] start OK", this.defaultMQProducer.getProducerGroup());
				this.serviceState = ServiceState.RUNNING;
				break;
			case RUNNING :
			case START_FAILED :
			case SHUTDOWN_ALREADY :
				throw new MQClientException("The producer service state not OK, maybe started once, "//
						+ this.serviceState//
						+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
			default :
				break;
		}
		// 11、立即调用MQClientInstance.sendHeartbeatToAllBrokerWithLock()方法，在该方法中，
		// 第一，向所有在MQClientInstance.brokerAddrTable列表中的Broker发送心跳消息；
		// 第二，向Filter过滤服务器发送REGISTER_MESSAGE_FILTER_CLASS请求码，更新过滤服务器中的Filterclass文件；
		this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
	}

	private void checkConfig() throws MQClientException {
		Validators.checkGroup(this.defaultMQProducer.getProducerGroup());

		if (null == this.defaultMQProducer.getProducerGroup()) {
			throw new MQClientException("producerGroup is null", null);
		}

		if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
			throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.", null);
		}
	}

	public void shutdown() {
		this.shutdown(true);
	}

	public void shutdown(final boolean shutdownFactory) {
		switch (this.serviceState) {
			case CREATE_JUST :
				break;
			case RUNNING :
				this.mQClientFactory.unregisterProducer(this.defaultMQProducer.getProducerGroup());
				if (shutdownFactory) {
					this.mQClientFactory.shutdown();
				}

				log.info("the producer [{}] shutdown OK", this.defaultMQProducer.getProducerGroup());
				this.serviceState = ServiceState.SHUTDOWN_ALREADY;
				break;
			case SHUTDOWN_ALREADY :
				break;
			default :
				break;
		}
	}

	@Override
	public Set<String> getPublishTopicList() {
		Set<String> topicList = new HashSet<String>();
		for (String key : this.topicPublishInfoTable.keySet()) {
			topicList.add(key);
		}

		return topicList;
	}

	@Override
	public boolean isPublishTopicNeedUpdate(String topic) {
		TopicPublishInfo prev = this.topicPublishInfoTable.get(topic);
		// 若遇到获取的prev对象为null或者该对象的messageQueueList:List<MessageQueue>列表变量为空则跳出该遍历过程
		return null == prev || !prev.ok();
	}

	@Override
	public TransactionCheckListener checkListener() {
		if (this.defaultMQProducer instanceof TransactionMQProducer) {
			TransactionMQProducer producer = (TransactionMQProducer) defaultMQProducer;
			return producer.getTransactionCheckListener();
		}

		return null;
	}

	@Override
	public void checkTransactionState(final String addr, final MessageExt msg, final CheckTransactionStateRequestHeader header) {
		Runnable request = new Runnable() {
			private final String brokerAddr = addr;
			private final MessageExt message = msg;
			private final CheckTransactionStateRequestHeader checkRequestHeader = header;
			private final String group = DefaultMQProducerImpl.this.defaultMQProducer.getProducerGroup();

			@Override
			public void run() {
				/**
				 * 4.1）获取DefaultMQProducerImpl对象的TransactionCheckListener值，
				 * 该值由应用层调用TransactionMQProducer.setTransactionCheckListener (TransactionCheckListener transactionCheckListener)方法设置；
				 * 在应用层实现TransactionCheckListener接口的checkLocalTransactionState方法，
				 * 该方法中检查本地的业务是否处理成功，根据处理结果来决定已发送到Broker的事务消息是应该提交还是回滚；
				 * 该方法返回LocalTransactionState. ROLLBACK_MESSAGE或者LocalTransactionState.COMMIT_MESSAGE或LocalTransactionState.UNKNOW；
				 */
				TransactionCheckListener transactionCheckListener = DefaultMQProducerImpl.this.checkListener();
				if (transactionCheckListener != null) {
					LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
					Throwable exception = null;
					try {
						localTransactionState = transactionCheckListener.checkLocalTransactionState(message);
					} catch (Throwable e) {
						log.error("Broker call checkTransactionState, but checkLocalTransactionState exception", e);
						exception = e;
					}

					this.processTransactionState(//
							localTransactionState, //
							group, //
							exception);
				} else {
					log.warn("checkTransactionState, pick transactionCheckListener by group[{}] failed", group);
				}
			}

			private void processTransactionState(//
					final LocalTransactionState localTransactionState, //
					final String producerGroup, //
					final Throwable exception) {
				// 4.2）构建EndTransactionRequestHeader对象
				final EndTransactionRequestHeader thisHeader = new EndTransactionRequestHeader();
				thisHeader.setCommitLogOffset(checkRequestHeader.getCommitLogOffset());
				thisHeader.setProducerGroup(producerGroup);
				// tranStateTableOffset变量等于收到到的消息的TranStateTableOffset值
				thisHeader.setTranStateTableOffset(checkRequestHeader.getTranStateTableOffset());
				// fromTransactionCheck变量等于true
				thisHeader.setFromTransactionCheck(true);
				thisHeader.setMsgId(message.getMsgId());
				thisHeader.setTransactionId(checkRequestHeader.getTransactionId());
				switch (localTransactionState) {
					case COMMIT_MESSAGE :
						thisHeader.setCommitOrRollback(MessageSysFlag.TransactionCommitType);
						break;
					case ROLLBACK_MESSAGE :
						thisHeader.setCommitOrRollback(MessageSysFlag.TransactionRollbackType);
						log.warn("when broker check, client rollback this transaction, {}", thisHeader);
						break;
					case UNKNOW :
						thisHeader.setCommitOrRollback(MessageSysFlag.TransactionNotType);
						log.warn("when broker check, client donot know this transaction state, {}", thisHeader);
						break;
					default :
						break;
				}

				String remark = null;
				if (exception != null) {
					remark = "checkLocalTransactionState Exception: " + RemotingHelper.exceptionSimpleDesc(exception);
				}

				try {
					// 4.3）向发送CHECK_TRANSACTION_STATE请求码的Broker发送END_TRANSACTION请求码，将该事务消息的处理结果告知Broker；
					DefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, thisHeader, remark, 3000);
				} catch (Exception e) {
					log.error("endTransactionOneway exception", e);
				}
			}
		};

		this.checkExecutor.submit(request);
	}

	@Override
	public void updateTopicPublishInfo(final String topic, final TopicPublishInfo info) {
		if (info != null && topic != null) {
			TopicPublishInfo prev = this.topicPublishInfoTable.put(topic, info);
			if (prev != null) {
				info.getSendWhichQueue().set(prev.getSendWhichQueue().get());
				log.info("updateTopicPublishInfo prev is not null, " + prev.toString());
			}
		}
	}

	@Override
	public boolean isUnitMode() {
		return this.defaultMQProducer.isUnitMode();
	}

	public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
		createTopic(key, newTopic, queueNum, 0);
	}

	public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
		this.makeSureStateOK();
		Validators.checkTopic(newTopic);

		this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
	}

	private void makeSureStateOK() throws MQClientException {
		if (this.serviceState != ServiceState.RUNNING) {
			throw new MQClientException("The producer service state not OK, "//
					+ this.serviceState//
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
		}
	}

	public List<MessageQueue> fetchPublishMessageQueues(String topic) throws MQClientException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().fetchPublishMessageQueues(topic);
	}

	public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
	}

	public long maxOffset(MessageQueue mq) throws MQClientException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
	}

	public long minOffset(MessageQueue mq) throws MQClientException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().minOffset(mq);
	}

	public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().earliestMsgStoreTime(mq);
	}

	public MessageExt viewMessage(String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		this.makeSureStateOK();

		return this.mQClientFactory.getMQAdminImpl().viewMessage(msgId);
	}

	public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws MQClientException, InterruptedException {
		this.makeSureStateOK();
		return this.mQClientFactory.getMQAdminImpl().queryMessage(topic, key, maxNum, begin, end);
	}

	/**
	 * DEFAULT ASYNC -------------------------------------------------------
	 */
	public void send(Message msg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
		send(msg, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
	}

	public void send(Message msg, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, InterruptedException {
		try {
			this.sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout);
		} catch (MQBrokerException e) {
			throw new MQClientException("unknown exception", e);
		}
	}

	private SendResult sendDefaultImpl(//
			Message msg, //
			final CommunicationMode communicationMode, //
			final SendCallback sendCallback, final long timeout//
	) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		// 校验 Producer 处于运行状态
		/**
		 * 1、检查DefaultMQProducerImpl的ServiceState是否为RUNNING，若不是RUNNING状态则直接抛出MQClientException异常给调用者
		 */
		this.makeSureStateOK();
		// 校验消息格式
		/**
		 * 2、校验Message消息对象的各个字段的合法性，其中Message对象的body的长度不能大于128KB
		 */
		Validators.checkMessage(msg, this.defaultMQProducer);

		final long maxTimeout = this.defaultMQProducer.getSendMsgTimeout() + 1000;
		final long beginTimestamp = System.currentTimeMillis();
		long endTimestamp = beginTimestamp;
		// 获取 Topic路由信息
		// 3、从topicPublishInfoTable: ConcurrentHashMap<String/* topic */, TopicPublishInfo>变量中获取TopicPublishInfo对象；
		TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

		/**
		 * 4、检查获取的TopicPublishInfo对象不为空，并且该对象的List<MessageQueue>队列也不为空
		 */
		if (topicPublishInfo != null && topicPublishInfo.ok()) {
			MessageQueue mq = null;// 最后选择消息要发送到的队列
			Exception exception = null;
			SendResult sendResult = null; // 最后一次发送结果
			int timesTotal = 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed();
			int times = 0; // 第几次发送
			String[] brokersSent = new String[timesTotal]; // 存储每次发送消息选择的broker名
			// 循环调用发送消息，直到成功
			/**
			 * 5、设置消息重试次数以及超时时间，
			 * 分别由参数DefaultMQProducer.retryTimesWhenSendFailed（默认为2）和DefaultMQProducer.sendMsgTimeout（默认为3000毫秒）设置；
			 * 重试次数为retryTimesWhenSendFailed+1，超时时间为sendMsgTimeout+1000；
			 * 当发送消息失败，在重试次数小于3和未到达4秒的超时时间，则再次发送消息
			 */
			for (; times < timesTotal && (endTimestamp - beginTimestamp) < maxTimeout; times++) {
				String lastBrokerName = null == mq ? null : mq.getBrokerName();
				// 6、选择发送的Broker和QueueId，即选择MessageQueue对象
				MessageQueue tmpmq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
				if (tmpmq != null) {
					mq = tmpmq;
					brokersSent[times] = mq.getBrokerName();
					try {
						// =========================7、调用发送消息核心方法=================
						sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
						endTimestamp = System.currentTimeMillis();
						switch (communicationMode) {
							case ASYNC :
								return null;
							case ONEWAY :
								return null;
							case SYNC :
								/**
								 * 8、对于同步方式发送消息，若未发送成功，
								 * 并且Producer设置允许选择另一个Broker进行发送（参数DefaultMQProducer. retryAnotherBrokerWhenNotStoreOK指定，默认为false，不允许），
								 * 则从第5步的检查发送失败次数和发送时间是否已经超过阀值开始重新执行；否则直接返回；
								 */
								if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
									if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
										continue;
									}
								}

								return sendResult;
							default :
								break;
						}
					} catch (RemotingException e) {
						log.warn("sendKernelImpl exception", e);
						log.warn(msg.toString());
						exception = e;
						endTimestamp = System.currentTimeMillis();
						continue;
					} catch (MQClientException e) {
						log.warn("sendKernelImpl exception", e);
						log.warn(msg.toString());
						exception = e;
						endTimestamp = System.currentTimeMillis();
						continue;
					} catch (MQBrokerException e) {
						log.warn("sendKernelImpl exception", e);
						log.warn(msg.toString());
						exception = e;
						endTimestamp = System.currentTimeMillis();
						switch (e.getResponseCode()) {
							case ResponseCode.TOPIC_NOT_EXIST :
							case ResponseCode.SERVICE_NOT_AVAILABLE :
							case ResponseCode.SYSTEM_ERROR :
							case ResponseCode.NO_PERMISSION :
							case ResponseCode.NO_BUYER_ID :
							case ResponseCode.NOT_IN_CURRENT_UNIT :
								continue;
							default :
								if (sendResult != null) {
									return sendResult;
								}

								throw e;
						}
					} catch (InterruptedException e) {
						log.warn("sendKernelImpl exception", e);
						log.warn(msg.toString());
						throw e;
					}
				} else {
					break;
				}
			} // end of for

			if (sendResult != null) {
				return sendResult;
			}

			String info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", //
					times, //
					(System.currentTimeMillis() - beginTimestamp), //
					msg.getTopic(), //
					Arrays.toString(brokersSent));

			info += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);

			throw new MQClientException(info, exception);
		}

		List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();
		if (null == nsList || nsList.isEmpty()) {
			throw new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null);
		}

		throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO), null);
	}

	private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
		TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
		if (null == topicPublishInfo || !topicPublishInfo.ok()) {
			// 在该变量中有该topic的记录，表示该Producer在向NameServer获取到topic信息之后会为该topic创建本地消息队列MessageQueue对象
			this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
			// 更新刚创建的TopicPublishInfo对象的变量值；
			// 对于第一次向NameServer获取该topic的TopicPublishInfo对象肯定是获取不到的，故该TopicPublishInfo对象的haveTopicRouterInfo等于false；
			/**
			 * 由Producer或Consumer端调用MQClientInstance. updateTopicRouteInfoFromNameServer()方法的目的主要有:
			 * 1）从NameServer获取该客户端订阅的每个topic的相关信息（包括每个topic对应的Broker信息和topic配置信息），用TopicRouteData对象表示；并将该对象信息存入MQClientInstance.topicRouteTable变量中；
			 * 2）根据返回的信息更新topic对应的BrokerName和BrokerId，保存在变量 MQClientInstance.brokerAddrTable中；
			 * 3）对于Prodcuer，根据每个topic以及配置的写队列个数创建MessageQueue消费队列，并存入变量DefaultMQProducerImpl.topicPublishInfoTable中；
			 * 4）对于已经订阅了该topic的Consumer，根据每个topic以及配置的读队列个数创建MessageQueue消费队列，并存入变量RebalanceImpl. topicSubscribeInfoTable中；
			 */
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
			topicPublishInfo = this.topicPublishInfoTable.get(topic);
		}

		// 检查返回TopicPublishInfo对象的haveTopicRouterInfo
		if (topicPublishInfo.isHaveTopicRouterInfo() || (topicPublishInfo != null && topicPublishInfo.ok())) {
			return topicPublishInfo;
		} else {
			// 获取默认topic值"TBW102"的TopicPublishInfo对象
			/**
			 * 在第一次向NameServer获取该topic的TopicPublishInfo对象时会出现此情况。
			 * 若Broker开启了自动创建Topic功能，则在启动Broker时会自动创建TBW102的主题，不建议开启此功能，而应该采用手工创建的方式创建TOPIC；
			 * why:
			 * 在Broker的配置文件中可以设置autoCreateTopicEnable参数为false或true，
			 * 该参数的用处：RocketMQ在发送消息时，会首先获取路由信息。
			 * 如果是新的消息，由于MQServer上面还没有创建对应的Topic，
			 * 这个时候，如果上面的配置打开的话，会返回默认TOPIC的（RocketMQ会在每台broker上面创建名为TBW102的TOPIC）路由信息，
			 * 然后Producer会选择一台Broker发送消息，选中的broker在存储消息时，发现消息的topic还没有创建，就会自动创建topic。
			 * *****************后果就是：以后所有该TOPIC的消息，都将发送到这台broker上，达不到负载均衡的目的。********************
			 * 所以建议关闭自动创建TOPIC的功能，即在配置文件中将autoCreateTopicEnable参数设置为false，
			 */
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
			topicPublishInfo = this.topicPublishInfoTable.get(topic);
			return topicPublishInfo;
		}
	}

	private SendResult sendKernelImpl(final Message msg, //
			final MessageQueue mq, //
			final CommunicationMode communicationMode, //
			final SendCallback sendCallback, //
			final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

		// 7.1====根据入参MessageQueue对象的brokerName从MQClientInstance.brokerAddrTable获取主用Broker（BrokerID=0）的地址；
		String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
		if (null == brokerAddr) {
			// 7.2 若该Broker地址为空，
			// 则再次调用tryToFindTopicPublishInfo(String topic)方法（详见第3步操作）从topicPublishInfoTable变量中获取TopicPublishInfo对象
			tryToFindTopicPublishInfo(mq.getTopic());
			// 然后在执行第7.1的操作获取主用Broker地址
			brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
		}

		SendMessageContext context = null;
		if (brokerAddr != null) {
			byte[] prevBody = msg.getBody();
			try {
				int sysFlag = 0;
				/**
				 * 7.3若消息内容大于规定的消息内容大小
				 * （由DefaultMQProducer.compressMsgBodyOverHowmuch参数指定，默认是4KB）
				 * 之后就使用java.util.zip.DeflaterOutputStream进行对消息内容进行压缩
				 */
				if (this.tryToCompressMessage(msg)) {
					// 7.4若消息内容经过压缩则置sysFlag标志位从右往左第1个字节为1
					sysFlag |= MessageSysFlag.CompressedFlag;
				}

				final String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
				if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
					// 若消息的property属性中TRAN_MSG字段不为空，并且是可解析为true的字符串，则将sysFlag标志位第3个字节为1；
					sysFlag |= MessageSysFlag.TransactionPreparedType;
				}

				/**
				 * 7.5）在应用层实现CheckForbiddenHook接口，
				 * 并可以调用DefaultMQProducerImpl.RegisterCheckForbiddenHook(CheckForbiddenHook checkForbiddenHook)方法来注册CheckForbiddenHook钩子，
				 * 可以实现CheckForbiddenHook接口，该接口的方法是在消息发送前检查是否属于禁止消息。
				 * 在此先检查是否注册CheckForbiddenHook钩子，若注册了则执行；
				 */
				if (hasCheckForbiddenHook()) {
					CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
					checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
					checkForbiddenContext.setGroup(this.defaultMQProducer.getProducerGroup());
					checkForbiddenContext.setCommunicationMode(communicationMode);
					checkForbiddenContext.setBrokerAddr(brokerAddr);
					checkForbiddenContext.setMessage(msg);
					checkForbiddenContext.setMq(mq);
					checkForbiddenContext.setUnitMode(this.isUnitMode());
					this.executeCheckForbiddenHook(checkForbiddenContext);
				}

				/**
				 * 7.6)在应用层实现SendMessageHook接口可以调用DefaultMQProducerImpl.registerSendMessageHook(SendMessageHook hook)方法注册SendMessageHook钩子，
				 * 实现该接口的类有两个方法，分别是消息发送前和消息发送后的调用。
				 * 在此先检查是否注册SendMessageHook钩子，
				 * 若注册了则执行sendMessageBefore方法；在发送结束之后在调用sendMessageAfter方法；
				 */
				if (this.hasSendMessageHook()) {
					context = new SendMessageContext();
					context.setProducerGroup(this.defaultMQProducer.getProducerGroup());
					context.setCommunicationMode(communicationMode);
					context.setBornHost(this.defaultMQProducer.getClientIP());
					context.setBrokerAddr(brokerAddr);
					context.setMessage(msg);
					context.setMq(mq);
					this.executeSendMessageHookBefore(context);
				}

				/**
				 * 7.7)构建SendMessageRequestHeader对象，
				 * 其中，该对象的defaultTopic变量值等于"TBW102", 
				 * defaultTopicQueueNums变量值等于DefaultMQProducer.defaultTopicQueueNums值，默认为4，
				 * queueId等于MessageQueue对象的queueId；
				 */
				SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
				requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
				requestHeader.setTopic(msg.getTopic());
				requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey()); // TBW102
				requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums()); // 4
				requestHeader.setQueueId(mq.getQueueId());
				requestHeader.setSysFlag(sysFlag);
				requestHeader.setBornTimestamp(System.currentTimeMillis());
				requestHeader.setFlag(msg.getFlag());
				requestHeader.setProperties(MessageDecoder.messageProperties2String(msg.getProperties()));
				requestHeader.setReconsumeTimes(0);
				requestHeader.setUnitMode(this.isUnitMode());
				if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
					String reconsumeTimes = MessageAccessor.getReconsumeTime(msg);
					if (reconsumeTimes != null) {
						requestHeader.setReconsumeTimes(new Integer(reconsumeTimes));
						MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_RECONSUME_TIME);
					}
				}

				/**
				 * 进行消息的发送；
				 * 在此方法中发送请求码为SEND_MESSAGE的RemotingCommand消息；
				 * 根据通信模式（communicationMode=同步[SYNC]、异步[ASYNC]、单向[ONEWAY]）分别调用不同的方法，将消息内容发送到Broker。
				 * 为了降低网络传输数量，设计了两种SendMessageRequestHeader对象，
				 * 一种是对象的变量名用字母简写替代，类名是SendMessageRequestHeaderV2，
				 * 一种是对象的变量名是完整的，类名是SendMessageRequestHeader。
				*A）同步[SYNC]：在调用发送消息的方法之后，同步等待响应消息，响应信息达到之后调用MQClientAPIImpl.processSendResponse(String brokerName, Message msg, RemotingCommand response)方法处理响应消息，返回给上层调用者； 
				*B）异步[ASYNC]：在发送消息之前，首先创建了内部匿名InvokeCallback类并实现operationComplete方法，并初始化ResponseFuture对象，其中InvokeCallback匿名类就是该对象的InvokeCallback变量值； 然后将该ResponseFuture对象以请求ID存入NettyRemotingAbstract.ResponseTable: ConcurrentHashMap<Integer==>opaque , ResponseFuture>变量中；最后在收到响应消息之后以响应ID（即请求ID）从NettyRemotingAbstract. ResponseTable变量中取ResponseFuture对象，然后调用InvokeCallback类的operationComplete方法，完成回调工作； 
				*C）单向[ONEWAY]：只负责将消息发送出去，不接受响应消息；
				 */
				SendResult sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(//
						brokerAddr, // 1
						mq.getBrokerName(), // 2
						msg, // 3
						requestHeader, // 4
						timeout, // 5
						communicationMode, // 6
						sendCallback// 7
				);

				if (this.hasSendMessageHook()) {
					context.setSendResult(sendResult);
					this.executeSendMessageHookAfter(context);
				}

				return sendResult;
			} catch (RemotingException e) {
				if (this.hasSendMessageHook()) {
					context.setException(e);
					this.executeSendMessageHookAfter(context);
				}
				throw e;
			} catch (MQBrokerException e) {
				if (this.hasSendMessageHook()) {
					context.setException(e);
					this.executeSendMessageHookAfter(context);
				}
				throw e;
			} catch (InterruptedException e) {
				if (this.hasSendMessageHook()) {
					context.setException(e);
					this.executeSendMessageHookAfter(context);
				}
				throw e;
			} finally {
				msg.setBody(prevBody);
			}
		}

		// 若brokerAddr仍然为空，抛出异常
		throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
	}

	private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));

	private boolean tryToCompressMessage(final Message msg) {
		byte[] body = msg.getBody();
		if (body != null) {
			if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {
				try {
					byte[] data = UtilAll.compress(body, zipCompressLevel);
					if (data != null) {
						msg.setBody(data);
						return true;
					}
				} catch (IOException e) {
					log.error("tryToCompressMessage exception", e);
					log.warn(msg.toString());
				}
			}
		}

		return false;
	}

	/**
	 * DEFAULT ONEWAY -------------------------------------------------------
	 */
	public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException {
		try {
			this.sendDefaultImpl(msg, CommunicationMode.ONEWAY, null, this.defaultMQProducer.getSendMsgTimeout());
		} catch (MQBrokerException e) {
			throw new MQClientException("unknow exception", e);
		}
	}

	/**
	 * KERNEL SYNC -------------------------------------------------------
	 */
	public SendResult send(Message msg, MessageQueue mq) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		return send(msg, mq, this.defaultMQProducer.getSendMsgTimeout());
	}

	public SendResult send(Message msg, MessageQueue mq, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		this.makeSureStateOK();
		Validators.checkMessage(msg, this.defaultMQProducer);

		if (!msg.getTopic().equals(mq.getTopic())) {
			throw new MQClientException("message's topic not equal mq's topic", null);
		}

		return this.sendKernelImpl(msg, mq, CommunicationMode.SYNC, null, timeout);
	}

	/**
	 * KERNEL ASYNC -------------------------------------------------------
	 */
	public void send(Message msg, MessageQueue mq, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
		send(msg, mq, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
	}

	public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, InterruptedException {
		this.makeSureStateOK();
		Validators.checkMessage(msg, this.defaultMQProducer);

		if (!msg.getTopic().equals(mq.getTopic())) {
			throw new MQClientException("message's topic not equal mq's topic", null);
		}

		try {
			this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, sendCallback, timeout);
		} catch (MQBrokerException e) {
			throw new MQClientException("unknow exception", e);
		}
	}

	/**
	 * KERNEL ONEWAY -------------------------------------------------------
	 */
	public void sendOneway(Message msg, MessageQueue mq) throws MQClientException, RemotingException, InterruptedException {
		this.makeSureStateOK();
		Validators.checkMessage(msg, this.defaultMQProducer);

		try {
			this.sendKernelImpl(msg, mq, CommunicationMode.ONEWAY, null, this.defaultMQProducer.getSendMsgTimeout());
		} catch (MQBrokerException e) {
			throw new MQClientException("unknow exception", e);
		}
	}

	/**
	 * SELECT SYNC -------------------------------------------------------
	 */
	public SendResult send(Message msg, MessageQueueSelector selector, Object arg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		return send(msg, selector, arg, this.defaultMQProducer.getSendMsgTimeout());
	}

	public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		return this.sendSelectImpl(msg, selector, arg, CommunicationMode.SYNC, null, timeout);
	}

	private SendResult sendSelectImpl(//
			Message msg, //
			MessageQueueSelector selector, //
			Object arg, //
			final CommunicationMode communicationMode, //
			final SendCallback sendCallback, final long timeout//
	) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		this.makeSureStateOK();
		Validators.checkMessage(msg, this.defaultMQProducer);

		TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
		if (topicPublishInfo != null && topicPublishInfo.ok()) {
			MessageQueue mq = null;
			try {
				mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
			} catch (Throwable e) {
				throw new MQClientException("select message queue throwed exception.", e);
			}

			if (mq != null) {
				return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
			} else {
				throw new MQClientException("select message queue return null.", null);
			}
		}

		throw new MQClientException("No route info for this topic, " + msg.getTopic(), null);
	}

	/**
	 * SELECT ASYNC -------------------------------------------------------
	 */
	public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
		send(msg, selector, arg, sendCallback, this.defaultMQProducer.getSendMsgTimeout());
	}

	public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, InterruptedException {
		try {
			this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, sendCallback, timeout);
		} catch (MQBrokerException e) {
			throw new MQClientException("unknown exception", e);
		}
	}

	/**
	 * SELECT ONEWAY -------------------------------------------------------
	 */
	public void sendOneway(Message msg, MessageQueueSelector selector, Object arg) throws MQClientException, RemotingException, InterruptedException {
		try {
			this.sendSelectImpl(msg, selector, arg, CommunicationMode.ONEWAY, null, this.defaultMQProducer.getSendMsgTimeout());
		} catch (MQBrokerException e) {
			throw new MQClientException("unknow exception", e);
		}
	}

	public TransactionSendResult sendMessageInTransaction(final Message msg, final LocalTransactionExecuter tranExecuter, final Object arg) throws MQClientException {
		if (null == tranExecuter) {
			throw new MQClientException("tranExecutor is null", null);
		}
		Validators.checkMessage(msg, this.defaultMQProducer);

		SendResult sendResult = null;
		// 将事务消息Message对象的properties属性中的“TRAN_MSG”字段设为true
		MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
		// PGROUP”字段设为该Producer的producerGroup值；
		MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
		try {
			// 将该事务消息发送到Broker
			sendResult = this.send(msg);
		} catch (Exception e) {
			throw new MQClientException("send message Exception", e);
		}

		LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
		Throwable localException = null;
		switch (sendResult.getSendStatus()) {
			case SEND_OK : {
				try {
					if (sendResult.getTransactionId() != null) {
						msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
					}
					// 调用业务层实现的executeLocalTransactionBranch方法，执行本地业务逻辑并返回本地事务状态LocalTransactionState；
					localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
					if (null == localTransactionState) {
						localTransactionState = LocalTransactionState.UNKNOW;
					}

					if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
						log.info("executeLocalTransactionBranch return {}", localTransactionState);
						log.info(msg.toString());
					}
				} catch (Throwable e) {
					log.info("executeLocalTransactionBranch exception", e);
					log.info(msg.toString());
					localException = e;
				}
			}
				break;
			case FLUSH_DISK_TIMEOUT :
			case FLUSH_SLAVE_TIMEOUT :
			case SLAVE_NOT_AVAILABLE :
				// 发送失败则不执行本地业务逻辑，直接将本地事务状态LocalTransactionState置为ROLLBACK_MESSAGE；
				localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
				break;
			default :
				break;
		}

		try {
			//
			this.endTransaction(sendResult, localTransactionState, localException);
		} catch (Exception e) {
			log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
		}

		TransactionSendResult transactionSendResult = new TransactionSendResult();
		transactionSendResult.setSendStatus(sendResult.getSendStatus());
		transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
		transactionSendResult.setMsgId(sendResult.getMsgId());
		transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
		transactionSendResult.setTransactionId(sendResult.getTransactionId());
		transactionSendResult.setLocalTransactionState(localTransactionState);
		return transactionSendResult;
	}

	private void endTransaction(//
			final SendResult sendResult, //
			final LocalTransactionState localTransactionState, //
			final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
		final MessageId id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
		String transactionId = sendResult.getTransactionId();
		final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
		// 创建EndTransactionRequestHeader对象，该对象的commitOrRollback变量是根据上一步的本地业务逻辑处理结果来设置
		// （可能为MessageSysFlag.TransactionCommitType/MessageSysFlag.TransactionRollbackTypeMessageSysFlag.TransactionNotType）、tranStateTableOffset变量值等于发送消息之后返回的该消息的queueoffset值、msgId变量值等于发送消息之后由Broker返回的MessageId；
		EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
		requestHeader.setTransactionId(transactionId);
		requestHeader.setCommitLogOffset(id.getOffset());
		switch (localTransactionState) {
			case COMMIT_MESSAGE :
				requestHeader.setCommitOrRollback(MessageSysFlag.TransactionCommitType);
				break;
			case ROLLBACK_MESSAGE :
				requestHeader.setCommitOrRollback(MessageSysFlag.TransactionRollbackType);
				break;
			case UNKNOW :
				requestHeader.setCommitOrRollback(MessageSysFlag.TransactionNotType);
				break;
			default :
				break;
		}

		requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
		requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
		requestHeader.setMsgId(sendResult.getMsgId());
		String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
		// 向Broker发送END_TRANSACTION请求码
		this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, this.defaultMQProducer.getSendMsgTimeout());
	}

	/**
	 * DEFAULT SYNC -------------------------------------------------------
	 */
	public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		return send(msg, this.defaultMQProducer.getSendMsgTimeout());
	}

	public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		// SYNC表示同步方式发送 timeout=3000
		return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
	}

	public ConcurrentHashMap<String, TopicPublishInfo> getTopicPublishInfoTable() {
		return topicPublishInfoTable;
	}

	public MQClientInstance getmQClientFactory() {
		return mQClientFactory;
	}

	public int getZipCompressLevel() {
		return zipCompressLevel;
	}

	public void setZipCompressLevel(int zipCompressLevel) {
		this.zipCompressLevel = zipCompressLevel;
	}

	public ServiceState getServiceState() {
		return serviceState;
	}

	public void setServiceState(ServiceState serviceState) {
		this.serviceState = serviceState;
	}
}
